package com.dtruth.dataflow;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;

import com.dtruth.dataflow.bolt.AutoChannelBolt;
import com.dtruth.dataflow.bolt.BodyReformatBolt;
import com.dtruth.dataflow.bolt.DataNormalBolt;
import com.dtruth.dataflow.bolt.DataPushOutBolt;
import com.dtruth.dataflow.bolt.DataRefererBolt;
import com.dtruth.dataflow.bolt.DataStatisticBolt;
import com.dtruth.dataflow.bolt.DataValidateBolt;
import com.dtruth.dataflow.bolt.PicsProcessBolt;
import com.dtruth.dataflow.bolt.RmDuplctBolt;
import com.dtruth.dataflow.spout.RabbitMQSpout;
import com.dtruth.dataflow.util.Env;
import com.dtruth.dataflow.util.MyLoggerFactory;

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;

public class DataFlowTopology {
	// static {System.setProperty("logback.configurationFile", "logback.xml");}

	private static Logger log = MyLoggerFactory
			.getLogger(DataFlowTopology.class);

	// private static Logger log =
	// LoggerFactory.getLogger(DataFlowTopology.class);

	public static void main(String[] args) throws InterruptedException,
			AlreadyAliveException {
		// Topology definition

		TopologyBuilder builder = new TopologyBuilder();

		// 从RabbitMQ中读取数据,并判断是否满足处理条件
		builder.setSpout("data-reader", new RabbitMQSpout());

		// 格式标准化：发布日期、域提取、分类重构
		builder.setBolt("data-normal", new DataNormalBolt()).shuffleGrouping(
				"data-reader");

		// 通过referer从redis中读取信息：频道映射及新闻质量分
		builder.setBolt("data-referer", new DataRefererBolt()).shuffleGrouping(
				"data-normal");

		// 自动频道划分：划分到频道
		builder.setBolt("auto-channel", new AutoChannelBolt()).shuffleGrouping(
				"data-referer");

		// Body标签替换
		builder.setBolt("body-reformat", new BodyReformatBolt())
				.shuffleGrouping("auto-channel");

		// Body标签替换
		builder.setBolt("data-validate", new DataValidateBolt())
				.shuffleGrouping("body-reformat");

		// 图片统计
		builder.setBolt("data-statistic", new DataStatisticBolt())
				.shuffleGrouping("data-validate");

		// 去重处理
		builder.setBolt("rm-duplicate", new RmDuplctBolt(), 4).shuffleGrouping(
				"data-validate");

		// 缩略图过滤
		builder.setBolt("pics-process", new PicsProcessBolt()).shuffleGrouping(
				"rm-duplicate");

		// 把结果推送到cms中
		builder.setBolt("data-pushout", new DataPushOutBolt(), 4)
				.shuffleGrouping("pics-process");

		// Configuration
		Config conf = new Config();

		conf.setDebug(true);

		// Topology run
		log.info("DataFlowTopology start");

		conf.put(Config.NIMBUS_HOST,
				Env.getInstance().getProperty("nimbus.host"));
		conf.put(
				Config.NIMBUS_THRIFT_PORT,
				Integer.valueOf(Env.getInstance().getProperty(
						"nimbus.thrift.port")));
		conf.setDebug(true);
		conf.setNumWorkers(Integer.valueOf(args[1]));
		try {
			StormSubmitter.submitTopology(args[0], conf,
					builder.createTopology());
		} catch (InvalidTopologyException e) {
			log.error(ExceptionUtils.getStackTrace(e));
		}

	}
}